package com.lefu.risk.storm.generdata;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.Map;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.utils.JedisPoolUtil;

public class ShuffGenerDataSpout extends BaseRichSpout {

	private static final long serialVersionUID = 1L;

	private SpoutOutputCollector collector;

	private BufferedReader bfr = null;
	int count = 0;
	long batch =0;

	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		try {
			bfr = new BufferedReader(new FileReader(new File("F:\\orderdata.txt")));
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		}
		this.collector = collector;
	}

	@Override
	public void nextTuple() {
		try {
			String readLine = bfr.readLine();
			if (readLine != null) {
				//collector.emit(new Values(readLine,redisUtil.getMicrosecond()/1000));
				
				collector.emit(new Values(readLine,System.currentTimeMillis()));
				
				if(++batch % 300 ==0)Thread.sleep(1);
			} else {
				bfr.close();
				if (count++ < 30)
					bfr = new BufferedReader(new FileReader(new File("F:\\orderdata.txt")));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("readLine","start"));
	}

}
